package com.shujia.card;

import com.shujia.bean.Car;
import com.shujia.common.FlinkTool;
import com.shujia.tf.CarMapper;
import com.shujia.tf.RealTimeCardWindowFlowToRedis;
import com.shujia.tf.RealTimeCardWindowProecss;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.time.Duration;
import java.util.Properties;

public class RealTimeCardWindowFlow extends FlinkTool {

    public static void main(String[] args) throws Exception {
        //创建flink的环境
        StreamExecutionEnvironment env =  FlinkTool.getFlinkEnv();

        env.setParallelism(1);


        //指定时间格式为事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //读取卡口过车数据

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092");
        properties.setProperty("group.id", "asdsadsa");

        //创建消费者

        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>(
                "cars",
                new SimpleStringSchema(),
                properties);


        //读取数据
        DataStream<String> carsDS = env.addSource(flinkKafkaConsumer);


        //解析数据
        DataStream<Car> carDS = carsDS.map(new CarMapper());


        //设置使劲字段和水位线
        WatermarkStrategy<Car> watermarkStrategy = WatermarkStrategy
                .<Car>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                .withTimestampAssigner((element, recordTimestamp) -> element.getTime() * 1000);

        SingleOutputStreamOperator<Car> wmDS = carDS.assignTimestampsAndWatermarks(watermarkStrategy);

        //取出卡口编号
        SingleOutputStreamOperator<Long> cardDS = wmDS.map(new MapFunction<Car, Long>() {
            @Override
            public Long map(Car value) throws Exception {
                return value.getCard();
            }
        });


        /*
         *
         * 1.2 按窗口统计每个卡扣流量 - 统计每个卡扣最近5分钟车流量，每隔1分钟统计一次
         */
        KeyedStream<Long, Long> longKeyedStream = cardDS.keyBy(card -> card);

        WindowedStream<Long, Long, TimeWindow> windowedStream = longKeyedStream.timeWindow(Time.seconds(5), Time.seconds(1));

        //统计每个窗口内的车流量
        SingleOutputStreamOperator<Tuple3<Long, Long, Long>> sumDS = windowedStream.process(new RealTimeCardWindowProecss());

        //将保存 到redis
        sumDS.addSink(new RealTimeCardWindowFlowToRedis());

        env.execute();


    }
}
